package SinkTest

import org.apache.flink.streaming.api.scala._
import SourceTest.Source01.SensorReading
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.client.Requests


object Elasticsearch_Sink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream1 = env.readTextFile("path")

    val httpHosts = new java.util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("localhost", 9200))

    val esSinkBuilder = new ElasticsearchSink.Builder[SensorReading](
      httpHosts,
      new ElasticsearchSinkFunction[SensorReading] {
        override def process(t: SensorReading, runtimeContext: RuntimeContext,
                             requestIndexer: RequestIndexer): Unit = {
          println("saving data: " + t)
          val json = new java.util.HashMap[String, String]()
          json.put("data", t.toString)
          val indexRequest =
            Requests.indexRequest().index("sensor").`type`("readingData").source(json)
          requestIndexer.add(indexRequest)
          println("saved successfully")
        }
      } )


//    stream1.addSink(esSinkBuilder.build())
  }

}
